#include "DataForward.h"
#include "NetMapSocket.h"
#include "NetPcapSocket.h"
#include "NetQtxSocket.h"
#include "TaskTime.h"
#include <iostream>

#define ETHER_MAX_LEN_CACHE  (ETHER_MAX_LEN - 80)

DataForward::DataForward(std::string srcmac, RawType type):
    Event(0),
    m_macstr(srcmac),
    m_type_io(type),
    m_raw_io(NULL),
    m_time_task(NULL)
{

}

DataForward::~DataForward()
{

}

int DataForward::init()
{
    try
    {
        switch (m_type_io)
        {
        case RawType::NETMAP:
            m_raw_io = new NetMapSocket(m_macstr);
            break;
        case RawType::PCAP:
            m_raw_io = new NetPcapSocket(m_macstr);
            break;
        case RawType::QTXSOCKET:
            m_raw_io = new NetQtxSocket(m_macstr);
            break;
        default:
            LOGFMTE("Type does not exist : %d ", static_cast<int>(m_type_io));
            return -1;
        }
    }
    catch (...)
    {
        LOGFMTE("init raw io fail : %s ", m_macstr.c_str());
        return -1;
    }
    
    m_time_task = new TaskTimer(rand() % 10000000);
    if (NULL == m_time_task || 0 != m_time_task->start())
    {
        delete m_time_task; m_time_task = NULL;
        delete m_raw_io; m_raw_io = NULL;
        return -1;
    }

    int ret = 0, tempValue[6];

    if (6 != (ret = sscanf(m_macstr.c_str(), "%02x:%02x:%02x:%02x:%02x:%02x",
        &tempValue[0], &tempValue[1], &tempValue[2], &tempValue[3], &tempValue[4], &tempValue[5]
    )))
    {
        delete m_time_task; m_time_task = NULL;
        delete m_raw_io; m_raw_io = NULL;
        LOGFMTE("failed to parse the mac : %s", m_macstr.c_str());
        return -1;
    }

    for (uint8_t i = 0; i < 6; ++i) *(m_src_mac() + i) = tempValue[i];

    if (0 != start())
    {
        delete m_time_task; m_time_task = NULL;
        delete m_raw_io; m_raw_io = NULL;
    
        LOGFMTE("data forward start thread fail : %s", m_macstr.c_str());
    
        return -1;
    }
    return 0;
}

int DataForward::shutdown()
{
    if (0 != stop())
    {
        LOGFMTE("stop thread fail : %s", m_macstr.c_str());
    }
    if (m_time_task)
    {
        m_time_task->stop();
    }
    delete m_time_task; m_time_task = NULL;
    delete m_raw_io; m_raw_io = NULL;

    return 0;
}

int DataForward::SendDataByAddr(MacType addr, Type type, DataCache data, int timeout)
{
    DataCache buff(ETHER_MAX_LEN + 1);
    NetPackHead header;
    int all_len, all_frame, last_frame;
    uint64_t taskid = m_time_task->GetDataId();

    memcpy(header.m_header.ether_dhost, addr(), sizeof(header.m_header.ether_dhost));
    memcpy(header.m_header.ether_shost, m_src_mac(), 6);
    header.m_base_cmd = TYPE_BASE_LAN;
    header.m_cmd = type;
    
    LOGFMTD("send to nic len : %d,type:%04X,remote_pnic:%s,remote_vic:%s,index:%u", 
        data.size(),(int)type,addr.toString().c_str(),MacType(reinterpret_cast<const unsigned char*>(data())).toString().c_str(),data.m_in_index);

    all_len = data.size();
    all_frame = (all_len + ETHER_MAX_LEN_CACHE - 1) / ETHER_MAX_LEN_CACHE;
    last_frame = all_len % ETHER_MAX_LEN_CACHE;

    for (int cur_frame = 1; cur_frame <= all_frame; ++cur_frame)
    {
        all_len = cur_frame == all_frame ? last_frame : ETHER_MAX_LEN_CACHE;
        header.m_curframe = cur_frame;
        header.m_allframe = all_frame;

        header.m_header.ether_type = ETHERTYPE_IP;
        header.m_data_len = all_len;
        header.m_task_id = taskid;

        header.toNet();

        memcpy(buff(), &header, sizeof(header));
        memcpy(buff() + sizeof(header), data() + ETHER_MAX_LEN_CACHE * (cur_frame - 1), all_len);
        buff.set(sizeof(header) + all_len);

        if (-1 == m_raw_io->SendData(buff, timeout))
        {
            LOGFMTD("send data error : task id : %lu, curframe : %d ,allframe : %d, len : %d",
                taskid, cur_frame, all_frame, all_len
            );
            return -1;
        }
        else
        {
            LOGFMTD("send data success : task id : %lu, curframe : %d ,allframe : %d, len : %d, data_len : %d",
                taskid, cur_frame, all_frame, buff.size(), all_len 
            );
        }
    }

    return 0;
}

int DataForward::VSubscribe(Type type, std::function<void(MacType, DataCache)> func)
{
    locker lock(m_mutex);
    m_func.insert({ type, func });

    return 0;
}

int DataForward::VUnsubscribe(Type type)
{
    auto iter = m_func.find(type);
    if (iter == m_func.end()) return -1;

    m_func.erase(iter);

    return 0;
}

void DataForward::CycleTask()
{
    DataCache  data(ETHER_MAX_LEN), newData;

    if (-1 == m_raw_io->RecvData(data, 1000))
    {
        auto iter = m_func.find(TYPE_ERROR);
        if (iter != m_func.end()) iter->second(MacType(), std::move(data));

        LOGFMTE("recv data error , exit !!!!");
        this->stop();

        return;
    }

    if (data.size() <= (int)sizeof(NetPackHead)) return;
    
    NetPackHead& header = *((NetPackHead*)data());
    header.toHost();

    if (header.m_header.ether_type != ETHERTYPE_IP ||header.m_base_cmd != TYPE_BASE_LAN) return;
    if (header.m_data_len <= 0) return;

    if (header.m_allframe < header.m_curframe || header.m_allframe < 0 || header.m_curframe < 0)
    {
        LOGFMTE("recv error data body, allframe : %u, curframe : %u", 
            header.m_allframe, header.m_curframe
        );
        return;
    }

    if (0 != ProcessPackage(newData, data)) return;

    {
        locker lock(m_mutex);
        auto iter = m_func.find((Type)header.m_cmd);
        if (iter == m_func.end()) return;

        iter->second(MacType(header.m_header.ether_shost), std::move(newData));
    }
}

int DataForward::ProcessPackage(DataCache& dstData, DataCache& srcData)
{
    NetPackHead& header = *((NetPackHead*)srcData());

    if (1 == header.m_allframe)
    {
        dstData.reset(header.m_data_len);
        memcpy(dstData(), srcData() + sizeof(NetPackHead), header.m_data_len);
        dstData.set(header.m_data_len);

        return 0;
    }
    
    TaskDataPtr _task = m_time_task->GetTask(header.m_task_id);
    if (NULL == _task)
    {
        _task = std::make_shared<TaskData>();
        _task->m_data = std::shared_ptr<unsigned char>(
            new uint8_t[header.m_allframe * ETHER_MAX_LEN_CACHE], [](unsigned char* p)
            {
                if (p) delete [] p;
                p = NULL;
            }
        );
        _task->m_taskid = header.m_task_id;

        m_time_task->AddTask(10000, _task);
    }

    uint64_t currRecvFrame = _task->m_status;
    currRecvFrame |= ((uint64_t)1 << (header.m_curframe - 1));

    if (currRecvFrame == _task->m_status) /* duplicate package */
    {
        return -1;
    }
    _task->m_status = currRecvFrame;

    if (header.m_allframe == header.m_curframe) _task->m_data_len = (header.m_allframe - 1) * ETHER_MAX_LEN_CACHE + header.m_data_len;

    memcpy(_task->m_data.get() + (header.m_curframe - 1) * ETHER_MAX_LEN_CACHE, srcData() + sizeof(NetPackHead), header.m_data_len);
    
    if (ALL_PACKAGE(_task->m_status, header.m_allframe)) /* recv all frame */
    {
        m_time_task->DeleteTask(header.m_task_id);

        dstData.reset(_task->m_data_len);
        dstData.set(_task->m_data_len);
        memcpy(dstData(), _task->m_data.get(), _task->m_data_len);

        return 0;
    }

    return 1;
}